﻿/*
Copyright (C) 2009  Tomasz Chrzanowski

This program is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.

This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
GNU General Public License for more details.

You should have received a copy of the GNU General Public License
along with this program.  If not, see <http://www.gnu.org/licenses/>
*/

using System;
using System.Collections.Generic;
using System.Data;
using System.Data.SqlClient;
using System.Diagnostics;
using System.Transactions;

namespace Com.TomCdc.ChangeDataCapture
{
    /// <summary>
    /// 
    /// </summary>
    public class CdcProcessor
    {
        #region Private fileds

        /// <summary>
        /// Default number of transactions to be processed per single <see cref="Process"/> method call.
        /// </summary>
        private const int DEF_TRANS_PER_CALL = 20;

        /// <summary>
        /// Initial capacity of the tables list Dictionary <see cref="_monitoredTables"/>.
        /// </summary>
        private const int TBL_LIST_INIT_CAPACITY = 10;

        /// <summary>
        /// Initial capacity of the columns list Dictionary <see cref="_monitoredColumns"/>. 
        /// </summary>
        private const int COL_LIST_INIT_CAPACITY = 500;

        /// <summary>
        /// The prefix of sql store procedure, which put together changes made to monitored table, and
        /// provide information whether columns' value has changed or not, after prefix comes related 
        /// table name
        /// </summary>
        private const string STATE_CHANGE_SP_PREFIX = "IdentifyChangedData__";

        /// <summary>
        /// The prefix of a column indicating whether column value has changed or not
        /// </summary>
        private const string STATE_CHANGE_COL_PREFIX = "isChanged__";

        /// <summary>
        /// The prefix of a column new value
        /// </summary>
        private const string STATE_CHANGE_COL_PREFIX_INS = "ins__";

        /// <summary>
        /// The prefix of a column old value
        /// </summary>
        private const string STATE_CHANGE_COL_PREFIX_DEL = "del__";

        /// <summary>
        /// The collection of table names which can be processed.
        /// </summary>
        private IDictionary<string, int> _monitoredTables;

        /// <summary>
        /// The collection of column names which can be processed.
        /// </summary>
        private Dictionary<int, Dictionary<string, int>> _monitoredColumns;

        /// <summary>
        /// The collection of tables and its columns oridinar positions.
        /// </summary>
        private Dictionary<string, Dictionary<int, string>> _sourceTableColPos;

        /// <summary>
        /// The sql connection string to the CDC stage database.
        /// </summary>
        private string _sqlConnStrCDCstage;

        /// <summary>
        /// The sql connection string to the CDC store database.
        /// </summary>
        private string _sqlConnStrCDCstore;

        /// <summary>
        /// The sql connection string to the CDC tracking database.
        /// </summary>
        private string _sqlConnStrCDCtrack;

        /// <summary>
        /// The number of the data change transactions to be processed in a single call 
        /// of <see cref="Process"/> method.
        /// </summary>
        private int _changeTransPerSqlTran;

        /// <summary>
        /// True if the object is initialized.
        /// </summary>
        private bool _isInitialized;

        /// <summary>
        /// The bit mask representing currently tracked columns of source table.
        /// </summary>
        private Dictionary<string, string> _tableTrackingBitMask;

        /// <summary>
        /// The collection of tracked source tables as <see cref="SourceTable"/> objects.
        /// </summary>
        private Dictionary<string, SourceTable> _sourceTableInfo;

        #endregion

        #region Public properties

        /// <summary>
        /// Gets or sets the number of the data change transactions to be processed in a single call 
        /// of <see cref="Process"/> method. More the change transactions to be processed per call, 
        /// longer lasting sql transaction to commit DB changes. Value has to be grater than 0 and 
        /// less then 100, if no default value is assigned.
        /// </summary>
        public int ChangeTransPerSqlTran
        {
            get
            {
                return _changeTransPerSqlTran;
            }

            set
            {
                _changeTransPerSqlTran = (value > 0 && value < 100 ? value : DEF_TRANS_PER_CALL);
            }
        }

        #endregion

        #region Public methods

        /// <summary>
        /// Initializes new instance of <see cref="CdcProcessor"/> class, 
        /// with default number of transactions per <see cref="Process"/> call.
        /// </summary>
        /// <param name="sqlConnStrCDCstage">The sql connection string to the CDC store database.</param>
        /// <param name="sqlConnStrCDCstore">The sql connection string to the CDC stage database.</param>
        /// <param name="sqlConnStrCDCtrack">The sql connection string to the CDC track database.</param>
        public CdcProcessor(string sqlConnStrCDCstage, string sqlConnStrCDCstore, string sqlConnStrCDCtrack)
        {
            // validates input arguments
            if (string.IsNullOrEmpty(sqlConnStrCDCstage))
            {
                throw new ArgumentException("The sqlConnStrCDCstage argument cannot be null or empty.");
            }

            if (string.IsNullOrEmpty(sqlConnStrCDCstore))
            {
                throw new ArgumentException("The sqlConnStrCDCstore argument cannot be null or empty.");
            }

            if (string.IsNullOrEmpty(sqlConnStrCDCtrack))
            {
                throw new ArgumentException("The sqlConnStrCDCtrack argument cannot be null or empty.");
            }

            // initializes Dictionaries for tables available for processing
            _monitoredTables = new Dictionary<string, int>(TBL_LIST_INIT_CAPACITY, StringComparer.OrdinalIgnoreCase);

            // initializes Dictionaries for columns available for processing
            // used TBL_LIST_INIT_CAPACITY capacity, because the first dimension od Dictionary stores
            // tables list
            _monitoredColumns = new Dictionary<int, Dictionary<string, int>>(TBL_LIST_INIT_CAPACITY);

            // initializes number of transactions being processed per Process() call
            _changeTransPerSqlTran = DEF_TRANS_PER_CALL;

            _sqlConnStrCDCstage = sqlConnStrCDCstage;
            _sqlConnStrCDCstore = sqlConnStrCDCstore;
            _sqlConnStrCDCtrack = sqlConnStrCDCtrack;

            _isInitialized = false;
        }

        /// <summary>
        /// Initializes all data required for processing start.
        /// </summary>
        private void Initialize()
        {
            // initializes list of tables which can be processed
            LoadMonitoredTables();

            // initializes list of columns which can be processed
            LoadMonitoredColumns();

            // mark object as initialized
            _isInitialized = true;
        }

        /// <summary>
        /// Performs processing of change transations in queue until queue empty
        /// </summary>
        /// <returns>Number of processed transactions.</returns>
        public int FullProcess()
        {
            int totalProcessedTrans = 0;
            int processedTrans = 0;

            // repeat until change transations queue empty
            while ((processedTrans = Process()) > 0)
            {
                totalProcessedTrans += processedTrans;
            }

            return totalProcessedTrans;
        }

        /// <summary>
        /// Performs processing of change transactions. The number of transactions processed per 
        /// single method call can be set by <see cref="CdcProcessor.ChangeTransPerSqlTran"/> property.
        /// </summary>
        /// <returns>Number of processed transactions.</returns>
        public int Process()
        {
            if (!_isInitialized)
            {
                Initialize();
            }

            Stopwatch T1 = Stopwatch.StartNew();

            // declares datatable collection to store processed data of changed columns of monitored source tables
            // each source table results are store in separate DataTable
            Dictionary<string, DataTable> changedData =
               new Dictionary<string, DataTable>(StringComparer.OrdinalIgnoreCase);


            // declares datatable to store transactions going to be processed from the staging database
            DataTable changeTranTable = new DataTable();

            // contains ID of the first transaction which is going to be processed
            Int64 fromTranId = Int64.MaxValue;

            // contains ID of the last transaction which is going to be processed
            Int64 toTranId = Int64.MinValue;

            // contains list of tables affected by data changes
            CaseInsensitiveList affectedTables = new CaseInsensitiveList();

            // contains command type of the data change (one of UPDATE, INSERT, DELETE)
            // this information is not available at the change transaction record from staging database,
            // but needed when inserting transaction change data to storage database
            IDictionary<Int64, string> tranCmdType = new Dictionary<Int64, string>();

            // creates scope for sql connection to staging database and sql command
            using (SqlCommand sqlCmd = new SqlCommand())
            using (SqlConnection sqlConnCDCstage = new SqlConnection(_sqlConnStrCDCstage))
            {
                // declares sql adapter which is used along the method
                SqlDataAdapter sqlAdapter = new SqlDataAdapter(sqlCmd);

                // changes parameters of sql command to retreive data change transactions from the satge database
                sqlCmd.CommandType = CommandType.StoredProcedure;
                sqlCmd.Connection = sqlConnCDCstage;
                sqlCmd.CommandTimeout = 60000;
                sqlCmd.CommandText = "SelectChangeTransactions";
                sqlCmd.Parameters.Clear();
                sqlCmd.Parameters.Add(new SqlParameter("@tranCount", _changeTransPerSqlTran));

                try
                {
                    sqlConnCDCstage.Open();

                    // fills DataTable with SP execution results
                    sqlAdapter.Fill(changeTranTable);
                }
                catch (SqlException ex)
                {
                    throw new CdcProcessorException("Database error occurred while" +
                        " selecting the change transactions from stage database. The time out might occure" +
                        " when changed data of source table are not commited for longer than 60 seconds.", ex);
                }

                // if no change transaction records to process finish processing
                if (changeTranTable.Rows.Count == 0)
                {
                    return 0;
                }

                // creates scope for transaction data table reader
                using (DataTableReader stageTranReader = changeTranTable.CreateDataReader())
                {
                    // indentifies column indexes
                    int idIdx = stageTranReader.GetOrdinal("ID");
                    int tableNameIdx = stageTranReader.GetOrdinal("tableName");

                    // goes through all change transactions and identifies affected tables 
                    // and min, max transaction ID
                    while (stageTranReader.Read())
                    {
                        fromTranId = Math.Min(fromTranId, stageTranReader.GetInt64(idIdx));
                        toTranId = Math.Max(toTranId, stageTranReader.GetInt64(idIdx));
                        // add table name to list if not already there, deleate for case insencitive check
                        if (!affectedTables.Contains(stageTranReader.GetString(tableNameIdx)))
                        {
                            affectedTables.Add(stageTranReader.GetString(tableNameIdx));
                        }
                    }
                }


                // changes paremeter list for next SP
                sqlCmd.Parameters.Clear();
                sqlCmd.Parameters.Add(new SqlParameter("@fromTranId", fromTranId));
                sqlCmd.Parameters.Add(new SqlParameter("@toTranId", toTranId));

                // iterates each table affected by the change
                foreach (string tableName in affectedTables)
                {
                    // procedure names have no spaces, if source table name contains spaces thay are removed in
                    // the procedure name
                    sqlCmd.CommandText = STATE_CHANGE_SP_PREFIX + tableName;

                    // each stored procedure in loop returns different column set (schema)
                    // so have to be reinitialized in every iteration
                    DataTable identifiedChanges = new DataTable();
                    try
                    {
                        sqlAdapter.Fill(identifiedChanges);
                    }
                    catch (SqlException ex)
                    {
                        throw new CdcProcessorException("Database error occurred while" +
                            " selecting the identified changed data of " + tableName +
                            " table from stage database. Timeout: " + sqlCmd.CommandTimeout, ex);
                    }


                    Debug.Assert(identifiedChanges.Rows.Count > 0,
                        "Executed SP has to return more then 0 records. ");

                    if (!_monitoredTables.ContainsKey(tableName))
                    {
                        throw new MissingInformationException("Transaction being processed involves table " +
                            "which is not specified in SourceTable table of storage database. " +
                            "Add table name to SourceTable and containing columns" +
                            " to SourceColumn table. Missing table name: " + tableName);
                    }


                    changedData.Add(tableName, GetChangeDataTable(tableName));

                    // extracts changed data and fills the changedData table
                    ProcessChangeStateTable(identifiedChanges, changedData[tableName],
                        _monitoredTables[tableName], tranCmdType);
                }
            }

            // write processed data to storage database
            PersistProcessingResult(changeTranTable, changedData, tranCmdType, fromTranId, toTranId);

            T1.Stop();
            Debug.WriteLine("[Process] exec time:" + T1.ElapsedMilliseconds);

            return changeTranTable.Rows.Count;
        }

        #endregion

        #region Private methods

        /// <summary>
        /// Initializes list of tables available for processing. The data is retrieved from the CdcStore database.
        /// List is stored in <see cref="Dictionary<string, int>"/> data structure. Table's database ID is stored as Dictionary value part.
        /// </summary>
        private void LoadMonitoredTables()
        {
            Stopwatch T = Stopwatch.StartNew();

            // creates scope for database connection
            using (IDbConnection sqlConn = new SqlConnection(_sqlConnStrCDCstore))
            {
                // prepares SP to pool tables list 
                IDbCommand sqlCmd = sqlConn.CreateCommand();
                sqlCmd.CommandText = "SelectSourceTables";
                sqlCmd.CommandType = CommandType.StoredProcedure;

                try
                {
                    // opens and executes SP
                    sqlConn.Open();
                    IDataReader sqlReader = sqlCmd.ExecuteReader();

                    // gets column indexes
                    int nameFieldIdx = sqlReader.GetOrdinal("Name");
                    int idFieldIdx = sqlReader.GetOrdinal("ID");

                    Debug.Assert(_monitoredTables.Count == 0);

                    // iterates through all records in the sqlreader
                    while (sqlReader.Read())
                    {
                        // adds table names (lower case, guaranteed by feeding store procedure) and IDs into list
                        _monitoredTables.Add(sqlReader.GetString(nameFieldIdx), sqlReader.GetInt32(idFieldIdx));
                    }

                    Debug.WriteLine("[ReadMonitoredTables] exec time:" + T.ElapsedMilliseconds);
                }
                catch (SqlException ex)
                {
                    throw new CdcProcessorException("Database error occured during tables list initialization.", ex);
                }
                catch (Exception ex)
                {
                    throw new CdcProcessorException("Error occured during tables list initialization.", ex);
                }
            }

        }


        /// <summary>
        /// Initializes list of columns available for processing. The data is retrieved from the CdcStore database.
        /// List is stored in <see cref="Dictionary<int, Dictionary<string, int>>"/> data structure. 
        /// First Dictionry's key refers to table ID the clolumns belong to, under its value is second Dictionary,
        /// which stores column name, ID pairs.
        /// </summary>
        private void LoadMonitoredColumns()
        {
            Stopwatch T = Stopwatch.StartNew();

            // creates scope for sql connections
            using (IDbConnection sqlConn = new SqlConnection(_sqlConnStrCDCstore))
            {
                // prepares SP to pool tables list 
                IDbCommand sqlCmd = sqlConn.CreateCommand();
                sqlCmd.CommandText = "SelectSourceCoulumns";
                sqlCmd.CommandType = CommandType.StoredProcedure;

                try
                {
                    // opens and executes SP
                    sqlConn.Open();
                    IDataReader sqlReader = sqlCmd.ExecuteReader();

                    // gets column indexes
                    int nameFieldIdx = sqlReader.GetOrdinal("Name");
                    int idFieldIdx = sqlReader.GetOrdinal("ID");
                    int tableIdFieldIdx = sqlReader.GetOrdinal("SourceTableId");

                    Debug.Assert(_monitoredColumns.Count == 0);

                    int tableId;

                    // iterates through all records in the sqlreader
                    while (sqlReader.Read())
                    {
                        tableId = sqlReader.GetInt32(tableIdFieldIdx);

                        // if no columns for met table added yet, add table ID and create Dictionary object
                        if (!_monitoredColumns.ContainsKey(tableId))
                        {
                            _monitoredColumns[tableId] =
                                new Dictionary<string, int>(COL_LIST_INIT_CAPACITY, StringComparer.OrdinalIgnoreCase);
                        }

                        // column names loaded into list as lower case, (guaranteed by feeding store procedure)
                        _monitoredColumns[tableId][sqlReader.GetString(nameFieldIdx)] = sqlReader.GetInt32(idFieldIdx);
                    }

                    Debug.WriteLine("[ReadMonitoredCoulumns] exec time:" + T.ElapsedMilliseconds);
                }
                catch (SqlException ex)
                {
                    throw new CdcProcessorException("Database error occured during columns list initialization.", ex);
                }
                catch (Exception ex)
                {
                    throw new CdcProcessorException("Error occured during columns list initialization.", ex);
                }
            }
        }

        /// <summary>
        /// The schema template to create DataTable to store processed data.
        /// </summary>
        private DataTable GetChangeDataTable(string tableName)
        {
            // validates input arguments
            if (string.IsNullOrEmpty(tableName))
            {
                throw new ArgumentException("The tableName argument cannot be null or empty.");
            }

            // initializes DataTable where schema to be built
            DataTable result = new DataTable(tableName);

            result.Columns.Add("ChangeTransactionId", typeof(Int64));
            result.Columns.Add("SourceRowId", typeof(Int64));
            result.Columns.Add("SourceColumnId", typeof(Int32));
            result.Columns.Add("SourceFieldValue", typeof(object));
            result.Constraints.Add("UniqueTranRowCol",
                new DataColumn[]{result.Columns["ChangeTransactionId"],
                                result.Columns["SourceRowId"],
                                result.Columns["SourceColumnId"]},
                true);

            return result;
        }

        /// <summary>
        /// Writes processed transactions to storage database and deletes processed transactions from 
        /// stage database. Operations performed in the scope of distributed transaction.
        /// </summary>
        /// <param name="changeTranTable">The DataTable containing transactions to save.</param>
        /// <param name="changedDataColl">The <code>DataTable</code> collection containing changed data of 
        /// source table. One <code>DataTable</code> per changed source table.</param>
        /// <param name="tranCmdType">The types of command performed in the change transaction.</param>
        /// <param name="fromTranId">The first change transaction which was processed.</param>
        /// <param name="toTranId">The last change transaction which was processed.</param>
        private void PersistProcessingResult(DataTable changeTranTable, IDictionary<string, DataTable> changedDataColl,
            IDictionary<Int64, string> tranCmdType, Int64 fromTranId, Int64 toTranId)
        {
            // validates input arguments
            if (changeTranTable == null)
            {
                throw new ArgumentNullException("changeTranTable");
            }

            if (fromTranId < 0 || fromTranId > toTranId)
            {
                throw new ArgumentOutOfRangeException("The fromTranId argument must be grater or equal 0 " +
                    "and lesser or equal toTranId argument.");
            }

            Debug.Indent();
            Stopwatch T1 = Stopwatch.StartNew();

            TransactionOptions transOpt = new TransactionOptions();

            // default isolation for distributed transaction is Serializable, ReadCommited is fine for our purpose
            transOpt.IsolationLevel = System.Transactions.IsolationLevel.ReadCommitted;

            // creates scope for distributed transaction, all sql connections opened in the scope will be automaticaly
            // enlisted in the transaction
#if !DEBUG
            using (TransactionScope distTranScope = new TransactionScope(TransactionScopeOption.RequiresNew, transOpt))
#endif
            {
                // first insert transactions data (referential integrity with ChangedData table,
                // so transactions must exist before loading data change records)
                PersistCdcStoreDbChangeTran(changeTranTable, tranCmdType);

                // inserts changed data of source tables
                foreach (DataTable changedData in changedDataColl.Values)
                {
                    // persists data one source table changes at a time
                    PersistCdcStoreDbChangedData(changedData);
                }

                // deletes processed transaction from staging database
                PersistCdcStageDbChanges(changeTranTable, fromTranId, toTranId);

#if !DEBUG
                // commits distributed transaction
                distTranScope.Complete();
#endif
            }

            Debug.WriteLine("[PersistProcessingResult] exec time:" + T1.ElapsedMilliseconds);
            Debug.Unindent();
        }

        /// <summary>
        /// Saves processed transactions to storage database.
        /// </summary>
        /// <param name="changeTranTable">The DataTable containing transactions to save.</param>
        /// <param name="tranCmdType">The Dictionary containing a transaction id as a key and transaction
        /// type (INSERT,UPDATE,DELETED) as a value.</param>
        private void PersistCdcStoreDbChangeTran(DataTable changeTranTable, IDictionary<Int64, string> tranCmdType)
        {
            if (changeTranTable == null)
            {
                throw new ArgumentNullException("changeTranTable");
            }

            Stopwatch T1 = Stopwatch.StartNew();

            // creats sql conection for storage database and open it
            using (SqlConnection sqlConnCDCstore = new SqlConnection(_sqlConnStrCDCstore))
            {
                // creates sql command
                SqlCommand sqlCmd = new SqlCommand("InsertTransactionRecord", sqlConnCDCstore);
                sqlCmd.CommandType = CommandType.StoredProcedure;
                sqlCmd.Parameters.Add(new SqlParameter("@ID", SqlDbType.BigInt));
                sqlCmd.Parameters.Add(new SqlParameter("@SourceTableId", SqlDbType.Int));
                sqlCmd.Parameters.Add(new SqlParameter("@TrackingMask", SqlDbType.VarChar));
                sqlCmd.Parameters.Add(new SqlParameter("@UserName", SqlDbType.VarChar));
                sqlCmd.Parameters.Add(new SqlParameter("@HostName", SqlDbType.VarChar));
                sqlCmd.Parameters.Add(new SqlParameter("@AppName", SqlDbType.VarChar));
                sqlCmd.Parameters.Add(new SqlParameter("@ChangeDate", SqlDbType.DateTime));
                sqlCmd.Parameters.Add(new SqlParameter("@Spid", SqlDbType.Int));
                sqlCmd.Parameters.Add(new SqlParameter("@CmdType", SqlDbType.VarChar));

                // creates object to keep state of async database insert operation
                IAsyncResult cmdARes = null;
                int rowsAffected;

                try
                {
                    sqlConnCDCstore.Open();

                    // goes through all processed transations from stage database
                    // and inserts to final storage database
                    foreach (DataRow tranRow in changeTranTable.Rows)
                    {
                        // populates procedure arguments
                        sqlCmd.Parameters["@ID"].Value = tranRow["ID"];
                        sqlCmd.Parameters["@SourceTableId"].Value = _monitoredTables[(string)tranRow["TableName"]];
                        sqlCmd.Parameters["@TrackingMask"].Value = GetTrackingBitMask((string)tranRow["TableName"]);
                        sqlCmd.Parameters["@UserName"].Value = tranRow["UserName"];
                        sqlCmd.Parameters["@HostName"].Value = tranRow["HostName"];
                        sqlCmd.Parameters["@AppName"].Value = tranRow["AppName"];
                        sqlCmd.Parameters["@ChangeDate"].Value = tranRow["ChangeDate"];
                        sqlCmd.Parameters["@Spid"].Value = tranRow["Spid"];
                        sqlCmd.Parameters["@CmdType"].Value = tranCmdType[(Int64)tranRow["ID"]];

                        // for first loop (IAsyncResult object is null) skip waiting for DB insert result
                        if (cmdARes != null)
                        {
                            rowsAffected = sqlCmd.EndExecuteNonQuery(cmdARes);
                            Debug.Assert(rowsAffected == 1);
                        }

                        // executes asyncronous DB insert operations
                        cmdARes = sqlCmd.BeginExecuteNonQuery();
                    }


                    Debug.Assert(cmdARes != null, "There have to be more than 0 change transactions to process," +
                        " and so object cannot be null.");

                    // ends SP execution of asyncronous command for last iteration of foreach loop
                    rowsAffected = sqlCmd.EndExecuteNonQuery(cmdARes);
                    Debug.Assert(rowsAffected == 1);

                    Debug.WriteLine("[PersistCdcStoreDbChangeTran] exec time:" + T1.ElapsedMilliseconds);
                }
                catch (SqlException ex)
                {
                    throw new CdcProcessorException("The database error occurred while inserting processed " +
                        "transactions to the store database." + ex.StackTrace, ex);
                }
                catch (Exception ex)
                {
                    throw new CdcProcessorException("The error occurred while inserting processed " +
                        "transactions to the store database." + ex.StackTrace, ex);
                }

            }

            T1.Stop();
            Debug.WriteLine("[PersistCdcStoreDbChangeTran] exec time:" + T1.ElapsedMilliseconds);
        }

        /// <summary>
        /// Saves processed changed data of tracked source table.
        /// </summary>
        /// <param name="changedData">The DataTable containing changed data.</param>
        private void PersistCdcStoreDbChangedData(DataTable changedData)
        {
            if (changedData == null)
            {
                throw new ArgumentNullException("changedData");
            }

            Stopwatch T1 = Stopwatch.StartNew(); ;

            // creats sql conection for storage database and open it
            using (SqlConnection sqlConnCDCstore = new SqlConnection(_sqlConnStrCDCstore))
            {
                // creates sql command
                SqlCommand sqlCmd = new SqlCommand();
                sqlCmd.CommandType = CommandType.StoredProcedure;
                sqlCmd.Connection = sqlConnCDCstore;

                // prepare SP name specific for the source table being processed
                sqlCmd.CommandText = "InsertChangeDataRecord__" + changedData.TableName;
                sqlCmd.Parameters.Add(new SqlParameter("@ChangeTransactionId", SqlDbType.BigInt));
                sqlCmd.Parameters.Add(new SqlParameter("@SourceRowId", SqlDbType.Int));
                sqlCmd.Parameters.Add(new SqlParameter("@SourceColumnId", SqlDbType.Int));
                sqlCmd.Parameters.Add(new SqlParameter("@SourceFieldValue", SqlDbType.Variant));

                // contains count of affected records by SP execution
                int rowsAffected;

                try
                {
                    sqlConnCDCstore.Open();

                    IAsyncResult cmdARes = null;

                    // goes through changed records and inserts into store DB
                    foreach (DataRow changeDataRow in changedData.Rows)
                    {
                        sqlCmd.Parameters["@ChangeTransactionId"].Value = changeDataRow["ChangeTransactionId"];
                        sqlCmd.Parameters["@SourceRowId"].Value = changeDataRow["SourceRowId"];
                        sqlCmd.Parameters["@SourceColumnId"].Value = changeDataRow["SourceColumnId"];
                        sqlCmd.Parameters["@SourceFieldValue"].Value = changeDataRow["SourceFieldValue"];

                        // for first loop (IAsyncResult object is null) no need to end SP execution
                        if (cmdARes != null)
                        {
                            // ends SP execution of asyncronous command
                            rowsAffected = sqlCmd.EndExecuteNonQuery(cmdARes);
                            Debug.Assert(rowsAffected == 1);
                        }

                        // executes asyncronous DB insert operations
                        cmdARes = sqlCmd.BeginExecuteNonQuery();
                    }

                    // ends SP execution, only if there where changed columns for processed change transactions
                    if (cmdARes != null)
                    {
                        // ends SP execution of asyncronous command for last iteration of foreach loop
                        rowsAffected = sqlCmd.EndExecuteNonQuery(cmdARes);
                        Debug.Assert(rowsAffected == 1);
                    }

                    Debug.WriteLine("[PersistCdcStoreDbChangedData] exec time:" + T1.ElapsedMilliseconds);
                }
                catch (SqlException ex)
                {
                    throw new CdcProcessorException("The database error occurred while inserting processed changed data " +
                        "to the store database." + ex.StackTrace, ex);
                }
                catch (Exception ex)
                {
                    throw new CdcProcessorException("The error occurred while inserting processed changed data " +
                        "to the store database." + ex.StackTrace, ex);
                }
            }
        }

        /// <summary>
        /// Deletes processed change transactions from the stage database.
        /// </summary>
        /// <param name="changeTranTable">The DataTable containing transactions to delete.</param>
        /// <param name="fromTranId">The begining of transactions range.</param>
        /// <param name="toTranId">The end of transactions range.</param>
        private void PersistCdcStageDbChanges(DataTable changeTranTable, Int64 fromTranId, Int64 toTranId)
        {
            // validates input arguments
            if (changeTranTable == null)
            {
                throw new ArgumentNullException("changeTranTable");
            }

            if (fromTranId < 0 || fromTranId > toTranId)
            {
                throw new ArgumentOutOfRangeException("The fromTranId argument must be grater or equal 0 " +
                    "and lesser or equal toTranId argument.");
            }

            Stopwatch T1 = Stopwatch.StartNew();

            // creates sql command
            SqlCommand sqlCmd = new SqlCommand();
            sqlCmd.CommandType = CommandType.StoredProcedure;

            // contains count of affected records by SP execution
            int rowsAffected;

            // creats sql conection for staging database and opens it
            using (SqlConnection sqlConnCDCstage = new SqlConnection(_sqlConnStrCDCstage))
            {
                try
                {
                    sqlConnCDCstage.Open();

                    // changes database to staging one
                    sqlCmd.Connection = sqlConnCDCstage;
                    sqlCmd.CommandText = "DeleteProcessedTransactions";
                    sqlCmd.Parameters.Clear();
                    sqlCmd.Parameters.Add(new SqlParameter("@fromTranId", fromTranId));
                    sqlCmd.Parameters.Add(new SqlParameter("@toTranId", toTranId));

                    // deletes processed transactions from staging database, tbChangeTran table has referencial
                    // integrity with tbSO_deleted and tbSO_inserted tables with on delete cascade attribute set,
                    // change data rows will be deleted automatically
                    rowsAffected = sqlCmd.ExecuteNonQuery();
                    Debug.Assert(rowsAffected == changeTranTable.Rows.Count);

                    Debug.WriteLine("[PersistCdcStageDbChanges] exec time:" + T1.ElapsedMilliseconds);
                }
                catch (SqlException ex)
                {
                    throw new CdcProcessorException("The database error occurred while deleting processed change " +
                        " transactions from the stage database.", ex);
                }
                catch (Exception ex)
                {
                    throw new CdcProcessorException("The error occurred while deleting processed change " +
                        " transactions from the stage database.", ex);
                }
            }
        }

        /// <summary>
        /// Proceses the change state table. The change state table contains record of all DML operations perfomed
        /// on the monitored table. One data row for each operation perfomed on the source row. The change row
        /// contains 3 columns per 1 source column. 
        /// <list type="bullet">
        /// <item>
        ///     <term>perfixed with isChanged__</term>
        ///     <description>comparison result of old and new value of a filed (1 differen, 0 equal)</description>
        /// </item>
        /// <item>
        ///     <term>prefixed with del__</term>
        ///     <description>old value of a filed</description>
        /// </item>
        /// <item>
        ///     <term>prefixed with ins__</term>
        ///     <description>new value of a field</description>
        /// </item>
        /// </list>
        /// Result of processing is saved to the processedState object which has structure of the table which will
        /// be final storage for all change data captured.
        /// </summary>
        /// <param name="changeState"></param>
        /// <param name="processedState"></param>
        /// <param name="sqlTran"></param>
        /// <param name="tableId"></param>
        /// <param name="tranId"></param>
        private void ProcessChangeStateTable(DataTable changeState, DataTable processedState, int tableId,
            IDictionary<Int64, string> tranCmdT)
        {
            // validates input arguments
            if (changeState == null)
            {
                throw new ArgumentNullException("changeState");
            }

            if (processedState == null)
            {
                throw new ArgumentNullException("processedState");
            }

            if (tranCmdT == null)
            {
                throw new ArgumentNullException("tranCmdT");
            }

            if (tableId < 1)
            {
                throw new ArgumentOutOfRangeException("The tableId argument must be grater than 0");
            }

            Stopwatch T1 = Stopwatch.StartNew();

            // declare columns of captured data changes, each source column has 3 fields of data,
            // a-difference b-inserted value c-deleted value
            DataColumnCollection cdcColumns = changeState.Columns;

            // declare list of all source table columns, containig struct with column name and its indexes in dataset
            // for each of 3 types of data mentioned above
            List<ColumnMeta> tableColumns = new List<ColumnMeta>();

            // use data table reader to travers through every record of captured change
            using (DataTableReader changeStateReader = changeState.CreateDataReader())
            {
                // index of uniqe identifier of the record in source table
                int srcTblRowIdIdx = changeStateReader.GetOrdinal("SourceRowId");

                // index of transaction id column
                int tranIdIdx = changeStateReader.GetOrdinal("TransactionId");

                // index of commad type column
                int cmdTypeIdx = changeStateReader.GetOrdinal("CmdType");

                // go through all columns and populate tableColumns list with columns indexes
                foreach (DataColumn prefixedColName in cdcColumns)
                {
                    // identify one type of column and extract oryginal column name
                    if (prefixedColName.ColumnName.StartsWith(STATE_CHANGE_COL_PREFIX))
                    {
                        // excract oryginal column name
                        string orgColName =
                            prefixedColName.ColumnName.Substring(STATE_CHANGE_COL_PREFIX.Length).ToLower();

                        if (!_monitoredColumns[tableId].ContainsKey(orgColName))
                        {
                            throw new MissingInformationException("Transaction being processed involves column " +
                                "which is not specified in tbMonitoredCoulumn table of storage database. " +
                                "Add related table name to tbMonitoredTable if needed, and missing column name " +
                                "to tbMonitoredColumn table. Missing column name: " + orgColName);
                        }

                        // having oryginal column name, identify indexes for all 3 column types
                        tableColumns.Add(new ColumnMeta(
                            // is new different than old (1-yes/0-no)
                            changeStateReader.GetOrdinal(STATE_CHANGE_COL_PREFIX + orgColName),
                            // old value (deleted)
                            changeStateReader.GetOrdinal(STATE_CHANGE_COL_PREFIX_DEL + orgColName),
                            // new value (inserted)
                            changeStateReader.GetOrdinal(STATE_CHANGE_COL_PREFIX_INS + orgColName),
                            // oryginal column name in source table
                            orgColName)
                            );
                    }
                }

                // contains ID of transaction currently being processed 
                Int64 changeTranId;

                // process all records of the source table data change
                while (changeStateReader.Read())
                {
                    changeTranId = changeStateReader.GetInt64(tranIdIdx);
                    tranCmdT[changeTranId] = changeStateReader.GetString(cmdTypeIdx);

                    // check if the column has changed if yes transpose data to storage dataset
                    foreach (ColumnMeta columnSet in tableColumns)
                    {
                        // check if processed column is changed
                        if (changeStateReader.GetBoolean(columnSet.diffColIdx))
                        {
                            // create new row with the schema of the dataset
                            DataRow dr = processedState.NewRow();
                            dr["ChangeTransactionId"] = changeTranId;
                            dr["SourceRowId"] = Convert.ToInt64(changeStateReader[srcTblRowIdIdx]);
                            dr["SourceColumnId"] = _monitoredColumns[tableId][columnSet.colName];

                            // if transaction command type DELETE get (new) field value from deleted column
                            // otherwise from inserted. Deleted columns will contain unchanged data since last
                            // insert/update, but we need complete record as final state of deleted record
                            if (tranCmdT[changeTranId] == "DELETE")
                            {
                                dr["SourceFieldValue"] = changeStateReader.GetValue(columnSet.delColIdx);
                            }
                            else
                            {
                                dr["SourceFieldValue"] = changeStateReader.GetValue(columnSet.insColIdx);
                            }

                            // add newly created record to recordset
                            processedState.Rows.Add(dr);
                        }
                    }
                }
            }

            Debug.WriteLine("[ProcessStateChangeTable] exec time:" + T1.ElapsedMilliseconds);
        }

        /// <summary>
        /// Gets the bit mask of currently tracked columns of source table.
        /// </summary>
        /// <param name="tableName">The name of a table.</param>
        /// <returns>The bit mask of tracked columns.</returns>
        private string GetTrackingBitMask(string tableName)
        {
            // inits the list of tables if not ready yet
            if (_sourceTableInfo == null)
            {
                LoadSourceTablesInfo();
                _tableTrackingBitMask = new Dictionary<string, string>(_sourceTableInfo.Count);
            }

            // checks if bit mask in the cache
            if (_tableTrackingBitMask.ContainsKey(tableName))
            {
                return _tableTrackingBitMask[tableName];
            }

            if (_sourceTableInfo.ContainsKey(tableName))
            {
                // gets the bit mask form SorceTable object
                _tableTrackingBitMask[tableName] = _sourceTableInfo[tableName].GetToActivateBitmaskStoreIds();
            }
            else
            {
                // tracking switched off on source table while still transactions to process on stage for that table exists
                // set no tracking mask "0x00"
                _tableTrackingBitMask[tableName] = "0x00";
            }
            return _tableTrackingBitMask[tableName];
        }

        private void LoadSourceTablesInfo()
        {
            // creates factory object to build source tables
            SourceSchemaFactory ssf = new SourceSchemaFactory(_sqlConnStrCDCtrack, _sqlConnStrCDCstore);

            ssf.InitializeFactory();

            // initializes SourceTable list object
            _sourceTableInfo = new Dictionary<string, SourceTable>(TBL_LIST_INIT_CAPACITY);
            
            // adds SourceTable objects to collection, only thoes being tracked
            foreach (SourceTable srcTable in ssf.CreateAllSourceTables(true))
            {
                if (srcTable.IsTracked)
                {
                    _sourceTableInfo[srcTable.TableIdentifier] = srcTable;
                }
            }            
        }

        #endregion
    }


}
